package com.swak.frame.ratelimit.interceptor;

import java.lang.reflect.Method;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

import javax.annotation.Resource;

import org.apache.commons.collections4.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.lang.Nullable;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Maps;
import com.swak.frame.cache.SwakLocalCache;
import com.swak.frame.common.TimeUnitConst;
import com.swak.frame.dto.base.BizScenario;
import com.swak.frame.exception.SwakAssert;
import com.swak.frame.exception.ThrowableWrapper;
import com.swak.frame.extension.executor.ExtensionExecutor;
import com.swak.frame.interceptor.AbstractBasicOperationSource;
import com.swak.frame.interceptor.OperationListener;
import com.swak.frame.ratelimit.LimitType;
import com.swak.frame.ratelimit.annotation.RateLimit;
import com.swak.frame.ratelimit.config.LimitCommand;
import com.swak.frame.ratelimit.config.LimitConfigVo;
import com.swak.frame.ratelimit.config.LimitOperation;
import com.swak.frame.ratelimit.event.RateLimitEvent;
import com.swak.frame.ratelimit.extension.RateLimitExtPtProvider;
import com.swak.frame.server.ServerRegisterService;
import com.swak.frame.system.SystemConfigurable;
import com.swak.frame.timer.CycleTask;
import com.swak.frame.timer.TimerTask;
import com.swak.frame.timer.WheelTimerHolder;
import com.swak.frame.ump.BizAlarmAdapter;

/**
 * Spring RateLimit operationSource
 * @ClassName: SpringRaterOperationSource.java
 * @author: colley.ma
 * @date: 2022/01/28
 */
public class SpringRaterOperationSource extends AbstractBasicOperationSource<LimitOperation>
		implements RaterOperationSource, ApplicationEventPublisherAware, SwakLocalCache {
	/**
	 * Canonical value held in cache to indicate no {@link RateLimit} annotation attribute was found for this method and we
	 * don't need to look again.
	 */
	private static final Collection<LimitOperation> NULL_ATTRIBUTE = Collections.emptyList();

	private final boolean publicMethodsOnly;

	private RaterAnnotationParserComposite annotationParser;

	private ApplicationEventPublisher publisher;

	@Resource
	private ExtensionExecutor extensionExecutor;

	private LimitType limitType = LimitType.LOCAL_TOKEN;

	@Autowired(required = false)
	private ServerRegisterService serverRegisterService;

	@Autowired(required = false)
	private BizAlarmAdapter bizAlarmAdapter;

	private boolean dynamic;

	@Resource
	private SystemConfigurable systemConfig;

	private RateLimitExtPtProvider rateLimitProvider;

	public SpringRaterOperationSource() {
		this(true);
	}

	public SpringRaterOperationSource(boolean publicMethodsOnly) {
		this.publicMethodsOnly = publicMethodsOnly;
	}

	/**
	 * Create a custom SpringRaterOperationSource.
	 * 
	 * @param annotationParser the LimitRaterAnnotationParser to use
	 */
	public SpringRaterOperationSource(RaterAnnotationParserComposite annotationParser) {
		this.publicMethodsOnly = true;
		SwakAssert.notNull(annotationParser, "DelegatingRateAnnotationParser must not be null");
		this.annotationParser = annotationParser;
	}

	@Override
	public Collection<LimitOperation> getBasicOperations(Method method, @Nullable Class<?> targetClass) {
		if (method.getDeclaringClass() == Object.class) {
			return null;
		}
		Object cacheKey = getCacheKey(method, targetClass);
		Collection<LimitOperation> cached = this.attributeCache.get(cacheKey);
		if (cached != null) {
			return (cached != NULL_ATTRIBUTE ? cached : null);
		} else {
			Collection<LimitOperation> operations = computeOperations(method, targetClass);
			if (CollectionUtils.isNotEmpty(operations)) {
				logger.trace("Adding @RateLimit method '" + method.getName() + "' with attribute: " + operations);
				this.attributeCache.put(cacheKey, operations);
				this.addListener(cacheKey);
			} else {
				this.attributeCache.put(cacheKey, NULL_ATTRIBUTE);
			}
			return operations;
		}
	}

	@Override
	protected boolean allowPublicMethodsOnly() {
		return this.publicMethodsOnly;
	}

	@Override
	protected Collection<LimitOperation> findAnnotationOperations(Class<?> clazz) {
		return annotationParser.parseAnnotations(clazz);
	}

	@Override
	protected Collection<LimitOperation> findAnnotationOperations(Method method,Class<?>  targetClass) {
		return annotationParser.parseAnnotations(method,targetClass);
	}

	@Override
	public Map<String, LimitOperation> allEntities() {
		Map<String, LimitOperation> entities = Maps.newHashMap();
		attributeCache.values().forEach(action -> {
			if (CollectionUtils.isNotEmpty(action)) {
				LimitOperation limit = action.iterator().next();
				entities.put(limit.getResource(), limit);
			}
		});
		return entities;
	}

	@Override
	public void afterSingletonsInstantiated() {
		if (rateLimitProvider == null) {
			this.rateLimitProvider = extensionExecutor.getExtensionPt(RateLimitExtPtProvider.class,
					BizScenario.valueOf(RateLimitExtPtProvider.RATE_LIMIT_BIZ_ID));
		}
		if(annotationParser==null) {
			this.annotationParser = beanFactory.getBean(RaterAnnotationParserComposite.class);
		}
	}

	@Override
	public void setApplicationEventPublisher(ApplicationEventPublisher publisher) {
		this.publisher = publisher;
	}

	public void publishEvent(LimitOperation operation) {
		publisher.publishEvent(
				new RateLimitEvent(this, LimitCommand.newLimitCommand(operation), limitType, operation.getResource()));
	}

	protected void addListener(Object cacheKey) {
		if (!this.attributeCache.containsKey(cacheKey)) {
			return;
		}
		Collection<LimitOperation> operations = this.attributeCache.get(cacheKey);
		if (CollectionUtils.isEmpty(operations)) {
			return;
		}
		LimitOperation operation = operations.iterator().next();
		this.publishEvent(operation);
		if (rateLimitProvider == null) {
			return;
		}
		// 如果rateLimitProvider不为空 需要监听更新
		OperationListener listener = () -> {
			TimerTask task = new CycleTask() {
				@Override
				public void invoke() throws ThrowableWrapper {
					watchListener(operation);
				}
			}.config(TimeUnitConst.MINUTES_1, TimeUnit.MINUTES, systemConfig.isRunJob());
			WheelTimerHolder.refreshableWheel().newTimeout(task, 1, TimeUnit.MINUTES);
		};
		listener.listener();
	}

	private void watchListener(LimitOperation operation) {
		if (rateLimitProvider == null) {
			return;
		}
		LimitConfigVo remoteSearch = rateLimitProvider.getConfig(operation.getResource());
		if (remoteSearch == null) {
			// 远程配置为空，并且配置未远程的设置为false
			if (operation.isRemote()) {
				operation.setRemote(false);
				operation.reset();// 重置为注解的限流方式,变为非动态计算QPS
				if (bizAlarmAdapter != null) {
					bizAlarmAdapter.alarmTag("ratelimit", "LimitRater config changed,config={}",
							JSON.toJSONString(operation));
				}
				this.publishEvent(operation);
			}
			// 如果是fisrt，需要pushlish
			if (operation.isFirst()) {
				operation.setFirst(false);
				this.publishEvent(operation);
			}
			return;
		}
		// 远程配置和本地QPS不一致，设置为远程的QPS
		int instances = 0;
		if (dynamic) {
			operation.setDynamic(dynamic);
			if (serverRegisterService != null) {
				instances = serverRegisterService.getInstances();
			}
		}
		boolean isNeedpublish = false;
		if (instances > 0 && !Objects.equals(operation.getInstances(), instances)) {
			if (bizAlarmAdapter != null) {
				bizAlarmAdapter.alarmTag("ratelimit",
						"LimitRater config instances changed,instances={},oldInstances={}", instances,
						operation.getInstances());
			}
			logger.warn("[swak-LimitRater] - LimitRater config instances changed,instances={},oldInstances={}",
					instances, operation.getInstances());
			isNeedpublish = true;
			operation.setInstances(instances);
		}
		if (!Objects.equals(operation.getQps(), remoteSearch.getQps())) {
			isNeedpublish = true;
			operation.setEnable(remoteSearch.getQps() > 0);
			operation.setQps(remoteSearch.getQps());
			operation.setCapacity(remoteSearch.getCapacity());
			operation.setRemote(true);
		}
		// 需要publishEvent
		if (isNeedpublish) {
			// 发布
			this.publishEvent(operation);
			if (bizAlarmAdapter != null) {
				bizAlarmAdapter.alarmTag("ratelimit", "LimitRater config changed,config:{}",
						JSON.toJSONString(operation));
			}
			logger.warn("[swak-LimitRater] - LimitRater config changed,config:{}", JSON.toJSONString(operation));
		}
	}

	public void setLimitType(LimitType limitType) {
		this.limitType = limitType;
	}

	public void setDynamic(boolean dynamic) {
		this.dynamic = dynamic;
	}

	@Override
	public boolean isCandidateClass(Class<?> targetClass) {
		return annotationParser.isCandidateClass(targetClass);
	}
}
